package cool.tsy.emos.wx.task;

import com.rabbitmq.client.*;
import cool.tsy.emos.wx.db.pojo.Message;
import cool.tsy.emos.wx.db.pojo.MessageRef;
import cool.tsy.emos.wx.exception.EmosException;
import cool.tsy.emos.wx.service.MessageService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Map;

/**
 * @author hongxuan.wang
 * @date Created in 2021/12/27 10:03
 * @description 消息任务
 * @mail 1443424326@qq.com
 */
@Component
@Slf4j
public class MessageTask {

    @Autowired
    private ConnectionFactory factory;

    @Autowired
    private MessageService messageService;

    /**
     * 同步发送消息
     *
     * @param topic   主题
     * @param message 消息对象
     */
    public void send(String topic, Message message) {
        // 向 MongoDB 保存消息数据，返回消息ID
        String id = messageService.insertMessage(message);
        // 向 RabbitMQ 发送消息
        try (Connection connection = factory.newConnection();
             Channel channel = connection.createChannel();
        ) {
            // 连接到某个 Topic
            channel.queueDeclare(topic, true, false, false, null);
            // 存放属性数据
            HashMap header = new HashMap();
            header.put("messageId", id);
            // 创建 AMQP 协议参数对象，添加附加属性
            AMQP.BasicProperties properties =
                    new AMQP.BasicProperties()
                            .builder()
                            .headers(header)
                            .build();
            channel.basicPublish("", topic, properties, message.getMsg().getBytes(StandardCharsets.UTF_8));
            log.debug("消息发送成功");
        } catch (Exception e) {
            log.error("执行异常", e);
            throw new EmosException("向MQ发送消息失败");
        }
    }

    /**
     * 异步发送消息
     *
     * @param topic   主题
     * @param message 消息对象
     */
    @Async
    public void sendAsync(String topic, Message message) {
        send(topic, message);
    }

    /**
     * 同步接收消息
     *
     * @param topic 主题
     * @return 接收消息数量
     */
    public int receive(String topic) {
        int i = 0;
        try (Connection connection = factory.newConnection();
             Channel channel = connection.createChannel();
        ) {
            // 从队列中获取消息，不自动确认
            channel.queueDeclare(topic, true, false, false, null);
            // Topic中有多少消息未知，所以使用死循环接收数据，直到接收不到消息，退出死循环
            while (true) {
                // 创建响应接收数据，禁止自动发送 Ack应答
                final GetResponse response = channel.basicGet(topic, false);
                if (response != null) {
                    AMQP.BasicProperties properties = response.getProps();
                    // 获取附加属性对象
                    Map<String, Object> headers = properties.getHeaders();
                    String messageId = headers.get("messageId").toString();
                    // 获取消息正文
                    byte[] body = response.getBody();
                    String message = new String(body);
                    log.debug("从RabbitMQ接收的消息：" + message);
                    MessageRef messageRef = new MessageRef();
                    messageRef.setMessageId(messageId);
                    messageRef.setReceiverId(Integer.parseInt(topic));
                    messageRef.setReadFlag(false);
                    messageRef.setLastFlag(true);
                    // 消息持久化到 MongoDB
                    messageService.insertRef(messageRef);
                    // 数据持久化后，才发送 ACK 应答，让 Topic 删除这条消息
                    long deliveryTag = response.getEnvelope().getDeliveryTag();
                    channel.basicAck(deliveryTag, false);
                    i++;
                } else {
                    // 接受不到消息，则退出死循环
                    break;
                }
            }
        } catch (Exception e) {
            log.error("执行异常", e);
        }
        return i;
    }

    /**
     * 异步接收消息
     *
     * @param topic 主题
     * @return 接收消息数量
     */
    @Async
    public int receiveAsync(String topic) {
        return receive(topic);
    }

    /**
     * 同步删除消息队列
     *
     * @param topic 主题
     */
    public void deleteQueue(String topic) {
        try (Connection connection = factory.newConnection();
             Channel channel = connection.createChannel();) {
            channel.queueDelete(topic);
            log.debug("消息队列成功删除");
        } catch (Exception e) {
            log.error("删除队列失败", e);
            throw new EmosException("删除队列失败");
        }
    }

    /**
     * 异步删除消息队列
     *
     * @param topic 主题
     */
    @Async
    public void deleteQueueAsync(String topic) {
        deleteQueue(topic);
    }
}
